package com.tsf.hbase;

import com.tsf.hbase.config.HBaseProperties;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Arrays;
import java.util.Map;

/**
 * hbase客户端初始化工厂
 *
 * @author taoningbo
 * @since 1.0
 */
public final class HBaseClientFactory implements HBaseClient {

    private static final Logger LOG = LoggerFactory.getLogger(HBaseClientFactory.class);
    private Configuration configuration = null;
    private final boolean initialized;

    /**
     * 自动初始化HBase配置
     *
     * @param properties 自定义hbase配置
     */
    private void autoInitConfiguration(HBaseProperties properties) {
        if (configuration == null) {
            configuration = HBaseConfiguration.create();
        }
        if (properties != null) {
            configuration.set("hbase.zookeeper.quorum", properties.getQuorum());
            configuration.set("hbase.master", properties.getMaster());
            configuration.set("hbase.zookeeper.property.clientPort", properties.getClientPort());
            configuration.set("hbase.client.write.buffer", properties.getWriteBuffer());
            Map<String, String> originProperties = properties.getProperties();
            if (!originProperties.isEmpty()) {
                for (String propertiesName : originProperties.keySet()) {
                    configuration.set(propertiesName, originProperties.get(propertiesName));
                }
            }
        }
    }

    /**
     * 构造一个{@link HBaseClientFactory}实例，你可以指定自定义tsf的hbase配置{@link HBaseProperties}。
     *
     * @param properties tsf定义的配置
     */
    public HBaseClientFactory(HBaseProperties properties) {
        this(null, properties);
    }

    /**
     * 构造一个{@link HBaseClientFactory}实例，你可以指定自定义原生配置{@link Configuration}和tsf的hbase配置{@link HBaseProperties}，
     * 另外，最终的有效配置优先级以原生配置为准。
     *
     * @param configuration hbase原生配置
     * @param properties    tsf定义的配置
     */
    public HBaseClientFactory(Configuration configuration, HBaseProperties properties) {
        if (configuration != null) {
            this.configuration = configuration;
        }
        if (properties == null) {
            throw new IllegalArgumentException("Cannot initialize HBase client without specifying HBaseProperties configuration!");
        }
        autoInitConfiguration(properties);
        if (properties.getTables() != null && !properties.getTables().isEmpty()) {
            LOG.info("The hbase data table is being created automatically through the data table properties specified in the configuration.");
            for (HBaseProperties.Table table : properties.getTables()) {
                autoCreateTable(table);
            }
        }
        initialized = true;
    }

    /**
     * 自动创建hbase数据表，只需要指定表的一些属性即可
     *
     * @param table 数据表的属性配置
     */
    public void autoCreateTable(HBaseProperties.Table table) {
        if (StringUtils.isEmpty(table.getTableName())) {
            throw new IllegalArgumentException("Auto created hbase table failed, htable name can not be null.");
        }
        TableName tableName = TableName.valueOf(table.getTableName());
        String[] familyNames = table.getFamilyNames();
        if (familyNames == null || familyNames.length == 0) {
            throw new IllegalArgumentException("Auto created hbase table failed, familyName can not be null or There is at least one family.");
        }
        try {
            Admin admin = connection().getAdmin();
            if (!admin.tableExists(tableName)) {
                LOG.info("Not exists this data table:{}, try to create a data table operator and try to set up some configuration.", tableName);
                HTableDescriptor desc = new HTableDescriptor(tableName)
                        .setCompactionEnabled(table.getCompactionEnabled())
                        .setConfiguration("hbase.table.sanity.checks", "false");
                for (String familyName : familyNames) {
                    HColumnDescriptor family = new HColumnDescriptor(familyName)
                            .setBlockCacheEnabled(table.getBlockCache())
                            .setBlocksize(table.getBlockSize())
                            .setBloomFilterType(table.getBloomFilterType())
                            .setTimeToLive(table.getTimeToLive())
                            .setCompressionType(table.getCompression());
                    desc.addFamily(family);
                }
                if (table.getSplitKeys() != null) {
                    byte[][] splitKeys = Arrays.stream(table.getSplitKeys().split(",")).map(Bytes::toBytes).toArray(byte[][]::new);
                    admin.createTable(desc, splitKeys);
                } else {
                    admin.createTable(desc);
                }
                LOG.info("Created data table succeed, tableName:{}|familyNames:{}", table.getTableName(), familyNames);
            } else {
                LOG.info("Exists this data table:{}, trying to get the data table operator and trying to modify some configuration.", table.getTableName());
                HTableDescriptor desc = admin.getTableDescriptor(tableName);
                if (desc != null) {
                    desc.setCompactionEnabled(table.getCompactionEnabled());
                    for (String familyName : familyNames) {
                        HColumnDescriptor family = desc.getFamily(Bytes.toBytes(familyName));
                        if (family != null) {
                            family.setBlockCacheEnabled(table.getBlockCache())
                                    .setBlocksize(table.getBlockSize())
                                    .setBloomFilterType(table.getBloomFilterType())
                                    .setTimeToLive(table.getTimeToLive())
                                    .setCompressionType(table.getCompression());
                            desc.modifyFamily(family);
                        } else {
                            family = new HColumnDescriptor(familyName)
                                    .setBlockCacheEnabled(table.getBlockCache())
                                    .setBlocksize(table.getBlockSize())
                                    .setBloomFilterType(table.getBloomFilterType())
                                    .setTimeToLive(table.getTimeToLive())
                                    .setCompressionType(table.getCompression());
                            desc.addFamily(family);
                        }
                    }
                    HColumnDescriptor[] families = desc.getColumnFamilies();
                    if (familyNames.length < families.length) {
                        for (HColumnDescriptor family : families) {
                            for (String familyName : familyNames) {
                                if (!family.getNameAsString().equals(familyName)) {
                                    desc.remove(familyName);
                                }
                            }
                        }
                    }
                    /*if (table.getSplitKeys() != null) {
                        admin.split(tableName, null);
                    }*/
                    admin.modifyTable(tableName, desc);
                }
                LOG.info("Modify data table succeed, tableName:{}|familyNames:{}", table.getTableName(), familyNames);
            }
        } catch (TableNotFoundException e) {
            LOG.warn("Auto created hbase table failed, data table not found.{}", e.getMessage());
        } catch (IOException e) {
            LOG.error("Auto created hbase table error.");
            throw new IllegalArgumentException("Auto created hbase table error", e);
        }
    }

    @Override
    public final Connection connection() {
        checkInitialized();
        Connection connection;
        try {
            connection = ConnectionFactory.createConnection(configuration);
        } catch (IOException e) {
            throw new IllegalArgumentException("Auto initialize hbase connection error.", e);
        }
        return connection;
    }

    /**
     * 转换一个{@code tableName}.
     *
     * @param tableName 表名称
     * @return TableName
     */
    @Override
    public final TableName getTableName(String tableName) {
        return TableName.valueOf(tableName);
    }

    /**
     * 根据{@code tableName}获取一个{@code Table}实例.
     *
     * @param tableName 表名称
     * @return Table
     */
    @Override
    public final Table getTable(String tableName) {
        try {
            return connection().getTable(getTableName(tableName));
        } catch (IOException e) {
            return null;
        }
    }

    public final void destroy() {
        checkInitialized();
        try {
            connection().close();
        } catch (IOException e) {
            LOG.warn("Closed hbase connection error.{}", e.getMessage());
        }
    }

    public void checkInitialized() {
        if (!initialized) {
            throw new IllegalArgumentException("'HBaseClientFactory' is not initialized, please " +
                    "initialize 'HBaseClientFactory' before using this method.");
        }
    }
}
